package com.surfilter.mq.rocketmq;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.surfilter.mq.util.ByteUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ProducerManage {
	
	private static Logger logger = LoggerFactory.getLogger(ProducerManage.class);
	
	private static Map<String, DefaultMQProducer> producers = new HashMap<String, DefaultMQProducer>();
	
	private static int maxProducerNum = 100;
	
	private static int defaultQueueNums = 16;
	
	/**
	 * 创建一个生产者，生产者个数最多不超过maxProducerNum个
	 * @param producerName
	 * @param groupName
	 * @param nameServerAddreses
	 * @throws MQClientException 
	 */
	public static void createProducer(String producerName, String groupName, String nameServerAddreses) throws MQClientException{
		//限制生产者个数
		if(producers.size() >= maxProducerNum){
			throw new IllegalArgumentException("生产者个数最多不允许超过" + maxProducerNum + "个");
		}
		//验证参数格式
		if(producerName == null || producerName.trim().equalsIgnoreCase("") || nameServerAddreses == null || nameServerAddreses.trim().equalsIgnoreCase("") || groupName == null || groupName.trim().equalsIgnoreCase("")){
			throw new IllegalArgumentException("参数异常，请提供有效参数！");
		}
		//添加生产者
		synchronized(producerName.intern()){
			if(!producers.containsKey(producerName)){
				DefaultMQProducer producer = new DefaultMQProducer(groupName);
//				producer.createTopic(key, newTopic, 16);
//				producer.createTopic(key, newTopic, queueNum);
				producer.setMaxMessageSize(10 * 1024 * 1024);
				producer.setNamesrvAddr(nameServerAddreses);
//				producer.setInstanceName(producerName);
				producer.start();
				producers.put(producerName, producer);
			}
		}
	}
	
	/**
	 * 创建一个生产者，生产者个数最多不超过maxProducerNum个
	 * @param producerName
	 * @param groupName
	 * @param nameServerAddreses
	 * @return 
	 * @throws MQClientException 
	 */
	public static boolean createProducerQueueNums(String topic, String producerName, String groupName, String nameServerAddreses, int queueNums) throws MQClientException{
		//限制生产者个数
		if(producers.size() >= maxProducerNum){
			throw new IllegalArgumentException("生产者个数最多不允许超过" + maxProducerNum + "个");
		}
		//验证参数格式
		if(producerName == null || producerName.trim().equalsIgnoreCase("") || nameServerAddreses == null || nameServerAddreses.trim().equalsIgnoreCase("") || groupName == null || groupName.trim().equalsIgnoreCase("")){
			throw new IllegalArgumentException("参数异常，请提供有效参数！");
		}
		//添加生产者
		synchronized(producerName.intern()){
			if(!producers.containsKey(producerName)){
				DefaultMQProducer producer = new DefaultMQProducer(groupName);
				producer.setMaxMessageSize(10 * 1024 * 1024);
				producer.setNamesrvAddr(nameServerAddreses);
//				producer.setInstanceName(producerName);
				producer.start();
				producers.put(producerName, producer);
				producer.createTopic(producer.getCreateTopicKey(), topic, queueNums);
			}
		}
		return true;
	}
	
	/**
	 * 
	 * @param producerName
	 * @param groupName
	 * @param nameServerAddreses
	 * @return
	 */
	public static boolean createProducerB(String producerName, String groupName, String nameServerAddreses){
		boolean result = false;
		try{
			createProducer(producerName, groupName, nameServerAddreses);
			result = true;
		}catch(Exception e){
			e.printStackTrace();
			logger.error("创建生产者:{}失败", producerName, e);
		}
		return result;
	}
	
	public static void multithreadSendMessage(final String producerName, final String topic, final String groupName, final String nameServerAddreses, List<Object> datas, int thread_num){
		List<List<? extends Object>> split_datas = splitData(datas, thread_num);
		if(split_datas == null || split_datas.isEmpty())return;
		for(final List<? extends Object> split_data : split_datas){
			new Thread(){
				@Override
				public void run() {
					String thread_producerName = producerName + Thread.currentThread().getName();
					String thread_groupName = groupName + Thread.currentThread().getName();
					try {
						boolean create_result = createProducerQueueNums(topic, thread_producerName, thread_groupName, nameServerAddreses, 16);
						if(!create_result)return;
						sendMessages(thread_producerName, topic, split_data);
					} catch (Exception e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
						logger.error("发送:{}数据异常:", thread_producerName, e);
					}
					deleteProducer(thread_producerName);
				}
			}.start();
		}
	}
	
	public static List<List<? extends Object>> splitData(List<? extends Object> datas, int split_num){
		List<List<? extends Object>> result = new ArrayList<>();
		if(datas == null || datas.isEmpty()) return result;
		int arg_num = (int)Math.round(datas.size()/Double.parseDouble(String.valueOf(split_num)));
		if(datas.size() <= arg_num){
			result.add(datas);
		}else{
			int all_num = 0;
			int result_num = 0;
			for(int  i = 0 ; i < datas.size(); i += arg_num){
				all_num ++ ;
				result_num = i;
				if(all_num >= split_num)break;
				result.add(datas.subList(i, i + arg_num));
			}
			if(result_num < datas.size()){
				result.add(datas.subList(result_num, datas.size()));
			}
		}
		return result;
	}
	
	public static List<List<? extends Object>> splitDataByMo(List<? extends Object> datas, int split_num){
		List<List<? extends Object>> results = new ArrayList<List<? extends Object>>();
		List<Object> result = new ArrayList<Object>(); 
		for(Object strs : datas){
			result.add(strs);
			if(result.size() % split_num == 0){
				results.add(result);
				result = new ArrayList<Object>();
			}
		}
		if(!result.isEmpty())results.add(result);
		return results;
	}
	
	/**
	 * 
	 * @param producerName
	 * @param topic
	 * @param datas
	 * @throws MQClientException
	 * @throws RemotingException
	 * @throws MQBrokerException
	 * @throws InterruptedException
	 */
	public static void sendMessage(String producerName, String topic, String tags, byte[] datas) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
		if(topic == null || datas == null || topic.trim().equalsIgnoreCase("") || datas.length == 0){
			throw new IllegalArgumentException("参数异常，请提供有效参数！");
		}
		synchronized(producerName.intern()){
			if(!producers.containsKey(producerName)){
				throw new IllegalArgumentException(producerName + "该生产者不存在，请先创建！");
			}
			producers.get(producerName).send(new Message(topic, tags, datas));
		}
	}
	
	/**
	 * 
	 * @param producerName
	 * @param topic
	 * @param datas
	 * @throws MQClientException
	 * @throws RemotingException
	 * @throws MQBrokerException
	 * @throws InterruptedException
	 */
	public static void sendMessage(String producerName, String topic, byte[] datas) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
		if(topic == null || datas == null || topic.trim().equalsIgnoreCase("") || datas.length == 0){
			throw new IllegalArgumentException("参数异常，请提供有效参数！");
		}
		synchronized(producerName.intern()){
			if(!producers.containsKey(producerName)){
				throw new IllegalArgumentException(producerName + "该生产者不存在，请先创建！");
			}
			producers.get(producerName).send(new Message(topic, datas));
		}
	}
	
	/**
	 * 批量发送消息
	 * @param producerName
	 * @param topic
	 * @param datas List集合中的元素必须可序列化
	 * @return 返回未发送完成的消息
	 * @throws MQClientException
	 * @throws RemotingException
	 * @throws MQBrokerException
	 * @throws InterruptedException
	 */
	public static List<Object> sendMessages(String producerName, String topic, String tags, List<Object> datas) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
		if(topic == null || datas == null || topic.trim().equalsIgnoreCase("") || datas.size() == 0){
			throw new IllegalArgumentException("参数异常，请提供有效参数！");
		}
		List<Object> result = new ArrayList<Object>();
		synchronized(producerName.intern()){
			if(!producers.containsKey(producerName)){
				throw new IllegalArgumentException(producerName + "该生产者不存在，请先创建！");
			}
			result.addAll(datas);
			try{
				for( int i = 0 ; i < datas.size() ; i ++){
					Object mesg = datas.get(i);
					boolean resu_bool = sendMessageB(producerName, topic, tags, ByteUtils.serialize(mesg));
					if(resu_bool)result.remove(mesg);
				}
			}catch(Exception e){
				logger.error("生产者:{}发送消息:{}失败", producerName, topic, e);
			}
		}
		return result;
	}
	
	/**
	 * 批量发送消息
	 * @param producerName
	 * @param topic
	 * @param datas List集合中的元素必须可序列化
	 * @return 返回未发送完成的消息
	 * @throws MQClientException
	 * @throws RemotingException
	 * @throws MQBrokerException
	 * @throws InterruptedException
	 */
	public static List<Object> sendMessages(String producerName, String topic, List<? extends Object> datas) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
		if(topic == null || datas == null || topic.trim().equalsIgnoreCase("") || datas.size() == 0){
			throw new IllegalArgumentException("参数异常，请提供有效参数！");
		}
		List<Object> result = new ArrayList<Object>();
		synchronized(producerName.intern()){
			if(!producers.containsKey(producerName)){
				throw new IllegalArgumentException(producerName + "该生产者不存在，请先创建！");
			}
			result.addAll(datas);
			try{
				for( int i = 0 ; i < datas.size() ; i ++){
					Object mesg = datas.get(i);
					boolean resu_bool = sendMessageB(producerName, topic, ByteUtils.serialize(mesg));
					if(resu_bool)result.remove(mesg);
				}
			}catch(Exception e){
				logger.error("生产者:{}发送消息:{}失败", producerName, topic, e);
			}
		}
		return result;
	}
	
	/**
	 * 
	 * @param producerName
	 * @param topic
	 * @param datas
	 * @return
	 */
	public static boolean sendMessageB(String producerName, String topic, byte[] datas){
		boolean result = false;
		try{
			sendMessage(producerName, topic, datas);
			result = true;
		}catch(Exception e){
			e.printStackTrace();
			logger.error("生产者:{}发送消息:{}失败", producerName, topic, e);
		}
		return result;
	}
	
	/**
	 * 
	 * @param producerName
	 * @param topic
	 * @param datas
	 * @return
	 */
	public static boolean sendMessageB(String producerName, String topic, String tags, byte[] datas){
		boolean result = false;
		try{
			sendMessage(producerName, topic, tags, datas);
			result = true;
		}catch(Exception e){
			e.printStackTrace();
			logger.error("生产者:{}发送消息:{}失败", producerName, topic, e);
		}
		return result;
	}
	
	/**
	 * @param producerName
	 */
	public static void deleteProducer(String producerName){
		if(producerName == null || producerName.trim().equalsIgnoreCase("")){
			throw new IllegalArgumentException("参数异常，请提供有效参数！");
		}
		synchronized(producerName.intern()){
			if(producers.containsKey(producerName)){
				producers.remove(producerName).shutdown();
			}
		}
	}
	
	public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, MQBrokerException, InterruptedException {
		String producerName = "producerName";
		String groupName = "groupName";
		String topic = "test_topic";
		String nameServerAddreses = "192.168.60.152:9876";
		ProducerManage.createProducer(producerName, groupName, nameServerAddreses);
		for(int i = 0 ; i < 10000 ; i ++){
			ProducerManage.sendMessage(producerName, topic, "2", new String("" + i).getBytes("utf-8"));
		}
		ProducerManage.deleteProducer(producerName);
	}
	
}
